#include "thread-pool.h"

// #define DEBUG_PROGRAM

void *workThreadWorkFunction(void *args);                           // 工作线程的工作函数

void *managerThreadWorkFunction(void *args);                        // 管理者线程的工作函数

int getAliveThreadNumber(ThreadPool *pool);                         // 获取线程池存活线程的数量

int getWorkThreadNumber(ThreadPool *pool);                          // 获取线程池工作线程的数量

long getWorkThreadAliveTime(ThreadPool *pool);                      // 获取工作线程的存活时间

void createWorkThread(ThreadPool *pool);                            // 创建工作线程

void workThreadExitFunction(ThreadPool *pool);                      // 工作线程退出函数

void refreshWorkThreadAliveTime(ThreadPool *pool);                  // 刷新工作线程的存活时间

struct ThreadPool {
    enum ThreadPoolState state;                 // 线程池状态

    enum TimeUnit time_unit;                    // 时间单位

    enum RejectionPolicy policy;                // 拒绝策略

    int keep_alive_time;                        // 线程存活时间(默认秒)

    int core_thread_number;                     // 核心线程数

    int max_thread_number;                      // 最大线程数

    TaskQueue *queue;                           // 任务队列;

    ThreadFactory *factory;                     // 线程工厂

    ThreadInfo *manager_thread_info;            // 管理者线程的 Info

    ThreadInfo **work_thread_info_list;         // 工作线程的 Info 列表

    int work_thread_number;                     // 工作线程数

    int alive_thread_number;                    // 存活线程数

    pthread_mutex_t task_queue_mutex;           // 任务队列的互斥锁

    pthread_mutex_t thread_pool_mutex;          // 线程池的互斥锁

    pthread_mutex_t work_thread_number_mutex;   // 工作线程数的互斥锁

    pthread_cond_t task_queue_is_fill_cond;     // 任务队列为满时的条件变量

    pthread_cond_t task_queue_is_empty_cond;    // 任务队列为空时的条件变量
};

ThreadPool *createThreadPool(int keep_alive_time, int core_thread_number, int max_thread_number, TaskQueue *queue,
                             ThreadFactory *factory, enum TimeUnit time_unit, enum RejectionPolicy policy) {
# ifdef DEBUG_PROGRAM
    printf("Start Create Thread Pool!\n");
# endif
    ThreadPool *pool = NULL;

    do {
        if (queue == NULL) {
# ifdef DEBUG_PROGRAM
            printf("Task Queue Is NULL!\n");
# endif
            break;
        }

        if (factory == NULL) {
# ifdef DEBUG_PROGRAM
            printf("Thread Factory Is NULL!\n");
# endif
            break;
        }

        pool = (ThreadPool *) malloc(sizeof(ThreadPool));
        if (pool == NULL) {
# ifdef DEBUG_PROGRAM
            printf("Thread Pool Apply Form Memory Fail!\n");
# endif
            break;
        }

        pool->work_thread_info_list = (ThreadInfo **) (ThreadInfo *) malloc(
                sizeof(ThreadInfo *) * max_thread_number);

        if (pool->work_thread_info_list == NULL) {
# ifdef DEBUG_PROGRAM
            printf("Work Thread Info List Apply From Memory Fail!\n");
# endif
            break;
        }

        for (int i = 0; i < max_thread_number; i++) {
            pool->work_thread_info_list[i] = NULL;
        }

        pool->state = RUNNING;

        pool->time_unit = time_unit;

        pool->policy = policy;

        pool->keep_alive_time = keep_alive_time;

        pool->core_thread_number = core_thread_number;

        pool->max_thread_number = max_thread_number;

        pool->queue = queue;

        pool->factory = factory;

        pool->work_thread_number = 0;

        pool->alive_thread_number = 0;

        pthread_mutex_init(&pool->task_queue_mutex, NULL);

        pthread_mutex_init(&pool->thread_pool_mutex, NULL);

        pthread_mutex_init(&pool->work_thread_number_mutex, NULL);

        pthread_cond_init(&pool->task_queue_is_fill_cond, NULL);

        pthread_cond_init(&pool->task_queue_is_empty_cond, NULL);

        // 创建管理者线程
        pool->manager_thread_info = createThread(factory, managerThreadWorkFunction, pool);

        // 创建工作线程
        for (int i = 0; i < core_thread_number; ++i) {
            createWorkThread(pool);
        }

# ifdef DEBUG_PROGRAM
        printf("Thread Pool Create Success!\n");
# endif
        return pool;

    } while (false);

    if (queue != NULL) {
        destroyTaskQueue(queue);
        queue = NULL;
    }

    if (factory != NULL) {
        destroyThreadFactory(factory);
        factory = NULL;
    }

    if (pool->work_thread_info_list != NULL) {
        free(pool->work_thread_info_list);
        pool->work_thread_info_list = NULL;
    }

    if (pool != NULL) {
        free(pool);
        pool = NULL;
    }

# ifdef DEBUG_PROGRAM
    printf("Thread Pool Create Failure!\n");
# endif
    return NULL;
}

void destroyThreadPool(ThreadPool *pool) {
# ifdef DEBUG_PROGRAM
    printf("Start Destroy Thread Pool!\n");
# endif

    // 将线程池的状态设置为关闭
    pthread_mutex_lock(&pool->thread_pool_mutex);
    pool->state = SHUTDOWN;
    pthread_mutex_unlock(&pool->thread_pool_mutex);

    // 回收管理者线程
    pthread_join(pool->manager_thread_info->pid, NULL);

    // 回收工作线程(工作线程在执行退出函数时, 会将线程池中的 Info 列表对于的 Info 设置为 NULL )
    int count = 0;
    bool is_complete_work = false;
    while (!is_complete_work && count++ != 100) {
        // 线程每间隔 300 毫秒就检测工作线程的状态
        struct timespec sleep_time = {0, 1000 * 1000 * 300};
        nanosleep(&sleep_time, NULL);

        is_complete_work = true;
        for (int i = 0; i < pool->max_thread_number; ++i) {
            ThreadInfo *info = pool->work_thread_info_list[i];
            if (info != NULL) {
                is_complete_work = false;
                pthread_cond_broadcast(&pool->task_queue_is_empty_cond);
                break;
            }
        }
    }

    // 超过半分钟的线程就让工作线程强制退出
    for (int i = 0; i < pool->max_thread_number; ++i) {
        ThreadInfo *info = pool->work_thread_info_list[i];
        if (info != NULL) {
            pthread_cancel(info->pid);
        }
    }

    // 销毁互斥锁和条件变量
    pthread_mutex_destroy(&pool->task_queue_mutex);
    pthread_mutex_destroy(&pool->thread_pool_mutex);
    pthread_cond_destroy(&pool->task_queue_is_fill_cond);
    pthread_cond_destroy(&pool->task_queue_is_empty_cond);

    // 销毁任务队列
    destroyTaskQueue(pool->queue);
    pool->queue = NULL;

    // 销毁任务工厂
    destroyThreadFactory(pool->factory);
    pool->factory = NULL;

    // 释放堆资源
    free(pool->work_thread_info_list);
    pool->work_thread_info_list = NULL;
    free(pool);
    pool = NULL;

# ifdef DEBUG_PROGRAM
    printf("Destroy Thread Pool Success!\n");
# endif
}

bool submitTask(ThreadPool *pool, callback work_function, void *args) {
    if (pool->state == RUNNING) {
        bool is_execute_task = false;
        Task *task = createTask(work_function, args);

        pthread_mutex_lock(&pool->task_queue_mutex);
        if (size(pool->queue) < capacity(pool->queue)) {
            push(pool->queue, task);
            // 唤醒工作线程执行任务
            pthread_cond_signal(&pool->task_queue_is_empty_cond);
# ifdef DEBUG_PROGRAM
            printf("Submit Task To Thread Pool\n");
# endif
        } else {
            // 根据线程池的拒绝策略执行对应的操作
            switch (pool->policy) {
                case ABORT:
# ifdef DEBUG_PROGRAM
                    printf("The Thread Pool Is Full, Throwing An Exception To The Upper Layer!\n");
# endif
                    break;

                case CALLER_RUN:
# ifdef DEBUG_PROGRAM
                    printf("The Thread Pool Is Full, Let The Task Committer Execute The Task!\n");
# endif
                    is_execute_task = true;
                    break;

                case BLOCKING_WAIT:
                    // 阻塞等待没有设置等待的超时时间
                    while (true) {
                        if (size(pool->queue) < capacity(pool->queue)) {
                            push(pool->queue, task);
                            // 唤醒工作线程执行任务
                            pthread_cond_signal(&pool->task_queue_is_empty_cond);
                            break;

                        } else {
# ifdef DEBUG_PROGRAM
                            printf("The Thread Pool Is Full, Blocking Waiting For The Task To Be Committed!\n");
# endif
                            pthread_cond_wait(&pool->task_queue_is_fill_cond, &pool->task_queue_mutex);
                        }
                    }
                    break;

                case DISCARD:
                    destroyTask(task);
                    task = NULL;
# ifdef DEBUG_PROGRAM
                    printf("The Thread Pool Is Full, Discard Current Task!\n");
# endif
                    break;

                case DISCARD_QUEUE_FRONT:
                    destroyTask(front(pool->queue));
                    pop(pool->queue);
                    push(pool->queue, task);
# ifdef DEBUG_PROGRAM
                    printf("The Thread Pool Is Full. Discard The Queue Front Task!\n");
# endif
                    break;
            }
        }
        pthread_mutex_unlock(&pool->task_queue_mutex);

        if (is_execute_task) {
            executeTask(task);
        }

        return true;
    }

# ifdef DEBUG_PROGRAM
    printf("The Thread Pool Has Been Closed And Cannot Continue To Submit Task!\n");
# endif
    return false;
}


void *workThreadWorkFunction(void *args) {
    ThreadPool *pool = (ThreadPool *) args;

# ifdef DEBUG_PROGRAM
    printf("Work Thread PID %p Create Success!\n", pthread_self());
# endif

    while (pool->state != TERMINATED) {
        pthread_mutex_lock(&pool->task_queue_mutex);
        while (empty(pool->queue)) {
# ifdef DEBUG_PROGRAM
            printf("Task Queue Size Is Empty, Work Thread Block Wait Signal!\n");
# endif
            pthread_cond_wait(&pool->task_queue_is_empty_cond, &pool->task_queue_mutex);
            pthread_mutex_lock(&pool->thread_pool_mutex);

            // 线程池为结束态时, 工作线程自行销毁
            // 线程池为关闭态时, 工作线程正常执行任务队列的任务, 但线程池不能提交任务
            if (pool->state == TERMINATED || (pool->state == SHUTDOWN && empty(pool->queue))) {
                pthread_mutex_unlock(&pool->thread_pool_mutex);
                pthread_mutex_unlock(&pool->task_queue_mutex);
                workThreadExitFunction(pool);
            }

            // 查看当前线程是否到了存活时间
            pthread_t pid = pthread_self();
            long current_time = time(NULL);
            long alive_time = getWorkThreadAliveTime(pool);
            // 超出存活时间的工作线程自行销毁
            if (alive_time + pool->keep_alive_time <= current_time &&
                pool->core_thread_number < pool->alive_thread_number) {
                pthread_mutex_unlock(&pool->thread_pool_mutex);
                pthread_mutex_unlock(&pool->task_queue_mutex);
                workThreadExitFunction(pool);
            }

            pthread_mutex_unlock(&pool->thread_pool_mutex);
        }

        // 从任务队列中获取任务
        Task *task = front(pool->queue);
        pop(pool->queue);
        pthread_mutex_unlock(&pool->task_queue_mutex);

        // 唤醒因为任务队列已满而休眠的线程
        pthread_cond_signal(&pool->task_queue_is_fill_cond);

        pthread_mutex_lock(&pool->work_thread_number_mutex);
        pool->work_thread_number++;
        pthread_mutex_unlock(&pool->work_thread_number_mutex);

        // 工作线程执行任务
        executeTask(task);

        pthread_mutex_lock(&pool->work_thread_number_mutex);
        pool->work_thread_number--;
        pthread_mutex_unlock(&pool->work_thread_number_mutex);

        // 刷新当前线程的存活时间
        refreshWorkThreadAliveTime(pool);

        // 销毁执行完成的任务
        destroyTask(task);
        task = NULL;
    }

    workThreadExitFunction(pool);
}

void *managerThreadWorkFunction(void *args) {
    ThreadPool *pool = (ThreadPool *) args;
# ifdef DEBUG_PROGRAM
    printf("Manager Thread PID %p Create Success!\n", pthread_self());
# endif

    while (pool->state == RUNNING) {
        // 通知超出存活时间的工作线程自行销毁
        bool is_notify = false;
        if (pool->core_thread_number < pool->alive_thread_number && empty(pool->queue)) {
            for (int i = 0; i < pool->max_thread_number; i++) {
                ThreadInfo *info = pool->work_thread_info_list[i];
                if (pool->work_thread_info_list[i] != NULL) {
                    long alive_time = info->alive_time;
                    long current_time = time(NULL);
                    if (alive_time + pool->keep_alive_time <= current_time) {
                        is_notify = true;
                    }
                }
            }
        }

        if (is_notify) { // 需要通知工作线程销毁时则不需要创建工作线程
            pthread_cond_broadcast(&pool->task_queue_is_empty_cond);

        } else {
            // 两种需要新建线程的情况:
            // 1.存活线程数小于线程池核心线程数
            // 2.任务队列已满且存活线程小于线程最大线程数
            if ((pool->alive_thread_number < pool->core_thread_number) ||
                (size(pool->queue) == capacity(pool->queue) &&
                 pool->alive_thread_number < pool->max_thread_number)) {

                pthread_mutex_lock(&pool->thread_pool_mutex);
                createWorkThread(pool);
                pthread_mutex_unlock(&pool->thread_pool_mutex);
            }
        }

        // 管理者线程每间隔 500 毫秒就进行线程池状态的检测
        struct timespec sleep_time = {0, 1000 * 1000 * 500};
        nanosleep(&sleep_time, NULL);
    }

# ifdef DEBUG_PROGRAM
    printf("Manager Thread PID %p Destroy Success!\n", pthread_self());
# endif
}

int getAliveThreadNumber(ThreadPool *pool) {
    return pool->alive_thread_number;
}

int getWorkThreadNumber(ThreadPool *pool) {
    return pool->work_thread_number;
}

long getWorkThreadAliveTime(ThreadPool *pool) {
    for (int i = 0; i < pool->max_thread_number; i++) {
        struct ThreadInfo *info = pool->work_thread_info_list[i];

        if (info != NULL && info->pid == pthread_self()) {
            return info->alive_time;
        }
    }

    return 0;
}

void createWorkThread(ThreadPool *pool) {
    pool->alive_thread_number++;
    ThreadInfo *info = createThread(pool->factory, workThreadWorkFunction, pool);

    for (int i = 0; i < pool->max_thread_number; i++) {
        if (pool->work_thread_info_list[i] == NULL) {
            pool->work_thread_info_list[i] = info;
            break;
        }
    }
}

void workThreadExitFunction(ThreadPool *pool) {
    pthread_t pid = pthread_self();

    pthread_mutex_lock(&pool->thread_pool_mutex);
    for (int i = 0; i < pool->max_thread_number; i++) {
        ThreadInfo *info = pool->work_thread_info_list[i];

        if (info != NULL && info->pid == pid) {
            free(info);
            pool->work_thread_info_list[i] = info = NULL;
        }
    }
    pthread_mutex_unlock(&pool->thread_pool_mutex);

# ifdef DEBUG_PROGRAM
    printf("Work Thread PID %p Destroy Success!\n", pthread_self());
# endif
    pthread_exit(NULL);
}

void refreshWorkThreadAliveTime(ThreadPool *pool) {
    pthread_t pid = pthread_self();

    for (int i = 0; i < pool->max_thread_number; i++) {
        ThreadInfo *info = pool->work_thread_info_list[i];

        pthread_mutex_lock(&pool->thread_pool_mutex);
        if (info != NULL && info->pid == pid) {
            info->alive_time = time(NULL);
            pthread_mutex_unlock(&pool->thread_pool_mutex);
            break;
        }
        pthread_mutex_unlock(&pool->thread_pool_mutex);
    }
}